/**
 * Copyright (C), 2011-2016 The Store
 * File Name: GenericEventListener.java
 * Encoding: UTF-8
 * Date: Feb 28, 2012
 * History: 
 */
package com.waynewan.qcomet.listener;

import java.io.IOException;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import javax.servlet.http.HttpServletResponse;

import org.apache.catalina.CometEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.waynewan.qcomet.event.ConnectionHolder;
import com.waynewan.qcomet.event.GenericEvent;
import com.waynewan.qcomet.message.MessageSender;

/**
 *
 * @author Wayne Wan(waynewan@mail.com)
 * @version Revision: 1.00 Date: Feb 28, 2012
 */
public class GenericEventListener extends AbstractListener {
	private static Log log = LogFactory.getLog(GenericEventListener.class);
	
	public static final String MSG_PREFIX = "[Q-COMET][EVENT:{0}][PAYLOAD:";
	public static final String MSG_POSTFIX = "][Q-COMET]";
	
	private ReadWriteLock lock ;
	private Lock readLock ;
	private Lock writeLock ;
	private Map<String, ConnectionHolder> connections = new HashMap<String, ConnectionHolder>();
	
	public GenericEventListener(ExecutorService threadPool){
		super(threadPool);
		lock = new ReentrantReadWriteLock();
		readLock = lock.readLock();
		writeLock = lock.writeLock();
	}
	
	/* (non-Javadoc)
	 * @see com.waynewan.qcomet.listener.QListener#updated(com.waynewan.qcomet.event.GenericEvent)
	 */
	@Override
	public void updated(GenericEvent event) {
		readLock.lock();
		try{
			StringBuffer sb = new StringBuffer();
			sb.append(MessageFormat.format(MSG_PREFIX, event.getEventName()));
			sb.append(event.getPayload());
			sb.append(MSG_POSTFIX);
			Set<String> keys = connections.keySet();
			// the key was the clientId(normally it was session id)
			for(String key : keys){
				if(event.getToken()!=null){
					if(key.equals(event.getToken())){
						log.debug("send a message to a specified client: " + key);
						ConnectionHolder conn = connections.get(key);
						conn.getSender().send(sb.toString());
						break;
					}
					continue;
				}
				log.debug("send a message to : " + key);
				ConnectionHolder conn = connections.get(key);
				conn.getSender().send(sb.toString());
			}
		}finally{
			readLock.unlock();
		}
	}

	/**
	 * add a connection(normally, the clientId was sessionId)
	 * @Title: addConn
	 * @Description:
	 * @param 
	 * @return void
	 * @author Wayne Wan(waynewan@mail.com)
	 * @date  Feb 29, 2012
	 */
	@Override
	public void addConn(String clientId, CometEvent event,  HttpServletResponse conn){
		writeLock.lock();
		try{
			if(connections.containsKey(clientId)) return;
			MessageSender sender =  new MessageSender(conn, this);
			threadPool.execute(sender);
			ConnectionHolder holder = new ConnectionHolder();
			holder.setClientId(clientId);
			holder.setSender(sender);
			holder.setEvent(event);
			connections.put(clientId, holder);
		}finally{
			writeLock.unlock();
		}
	}
	
	/**
	 * remove a connection(stop message sender, disconnect client, remove connection from context)<br>
	 * normally the clientId was sessionId
	 * @Title: removeConn
	 * @Description:
	 * @param 
	 * @return void
	 * @author Wayne Wan(waynewan@mail.com)
	 * @date  Feb 29, 2012
	 */
	@Override
	public void removeConn(String clientId){
		writeLock.lock();
		try{
			ConnectionHolder conn = connections.get(clientId);
			if(conn!=null){
				conn.getSender().stop();
				try {
					conn.getEvent().close();
				} catch (IOException e) {
					log.error("Close connection failed!", e);
				}
			}
			connections.remove(clientId);
		}finally{
			writeLock.unlock();
		}
	}

	@Override
	public int getConnSize() {
		return connections.size();
	}
}
